package cn.cihon.ubi.qorosauto.storm;
import java.util.Map;  

import redis.clients.jedis.Jedis;

import cn.cihon.ubi.qorosauto.bo.TripData;
import cn.cihon.ubi.qorosauto.dao.redis.TripDataToRedis;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
  
public class RedisStorageBolt extends BaseRichBolt{  
  
	/**
	 * 把计算后的行程数据存储到Redis中  
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector =null;
	private TripData tripData = null;
	private TripDataToRedis tripDataToRedis = null;
	private Jedis jedis = null;

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
		this.tripData = new TripData();
		this.tripDataToRedis = new TripDataToRedis();
		jedis = getJedis();

	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		tripData = (TripData) input.getValue(0);
		System.out.println("2.redis，开始存储");
		tripDataToRedis.store(tripData,jedis);
		this.collector.ack(input);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("Redisdata"));
		
	}
	
	public  Jedis getJedis()        //从Jedis连接池中获取jedis对象
	 {
		// jedis = JedisUtil.getPool().getResource();
		 jedis = new Jedis("123.57.54.113", 6379);
		 return jedis;
	 }
	
}